package cn.dfun.sample.flink.udftest

import cn.dfun.sample.flink.apitest.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.functions.{ScalarFunction, TableFunction}
import org.apache.flink.table.api.scala._
import org.apache.flink.types.Row

/**
  * 自定义函数
  * 对字段拆分计算长度
  */
object TableFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val settings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val tableEnv = StreamTableEnvironment.create(env, settings)
    val inputPath = "C:\\wor\\flink-sample\\src\\main\\resources\\sensor"
    val inputStream= env.readTextFile(inputPath)

    val dataStream = inputStream
      .map(data => {
        var arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000L
      })
    val sensorTable = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timestamp.rowtime as 'ts)

    val split = new Split("_")
    val resultTable = sensorTable
      // 侧向连接 笛卡尔积
      .joinLateral(split('id) as ('word, 'length))
      .select('id, 'ts, 'word, 'length)
    // sql
    tableEnv.createTemporaryView("sensor", sensorTable)
    tableEnv.registerFunction("split", split)
    val resultSqlTable = tableEnv.sqlQuery(
      """
        |select
        |id, ts, word, length
        |from sensor, lateral table(split(id)) as splitid(word, length)
      """.stripMargin)
    resultTable.toAppendStream[Row].print("result")
    resultSqlTable.toAppendStream[Row].print("sql")
    env.execute("table function test")
  }
}

// 自定义TableFunction
class Split(seperator: String) extends TableFunction[(String, Int)] {
  def eval(str: String): Unit = {
    str.split(seperator).foreach(
      word => collect((word, word.length))
    )
  }
}
